package com.zhang.flink.example;

import com.zhang.flink.bean.ApacheLog;
import com.zhang.flink.bean.UrlViewCountWindow;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;

/**
 * @title: 每隔5秒，输出最近10分钟内访问量最多的前N个URL
 * @author: zhang
 * @date: 2022/3/31 22:36
 */
public class HotUrlCountTopN {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .readTextFile("/Users/apple/project-git/flink-example/src/main/resources/apache.log")
                //.socketTextStream("localhost",9999)
                .map(new RichMapFunction<String, ApacheLog>() {
                    private SimpleDateFormat sdf;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        sdf = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");
                    }

                    @Override
                    public ApacheLog map(String value) throws Exception {
                        String[] fields = value.split(" ");
                        return new ApacheLog(
                                fields[0],
                                fields[1],
                                sdf.parse(fields[3]).getTime(),
                                fields[5],
                                fields[6]
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<ApacheLog>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                                .withTimestampAssigner(new SerializableTimestampAssigner<ApacheLog>() {
                                    @Override
                                    public long extractTimestamp(ApacheLog element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                )
                .keyBy(data -> data.url)
                .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.seconds(5)))
                .allowedLateness(Time.minutes(1))
                .sideOutputLateData(new OutputTag<ApacheLog>("apache-log") {
                })
                //求出每个url在每个窗口的pv
                .aggregate(new CountAgg(), new WindowResult())
                .keyBy(data -> data.windowEndTime)
                .process(new ProcessFunc(3))
                .print();

        env.execute();
    }

    private static class ProcessFunc extends KeyedProcessFunction<Long, UrlViewCountWindow, String> {
        private int n;

        public ProcessFunc(int n) {
            this.n = n;
        }

        private MapState<String, UrlViewCountWindow> mapState;

        @Override
        public void open(Configuration parameters) throws Exception {
            mapState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<String, UrlViewCountWindow>(
                            "map-state",
                            Types.STRING,
                            Types.POJO(UrlViewCountWindow.class)
                    )
            );
        }

        @Override
        public void processElement(UrlViewCountWindow value, KeyedProcessFunction<Long, UrlViewCountWindow, String>.Context ctx, Collector<String> out) throws Exception {
            mapState.put(value.url, value);

            //触发定时器输出统计结果，但是保留状态信息，因为要等待迟到数据
            ctx.timerService().registerEventTimeTimer(value.windowEndTime + 1L);
            //允许最大迟到时间到达，清空状态
            ctx.timerService().registerEventTimeTimer(value.windowEndTime + 63001L);
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<Long, UrlViewCountWindow, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            if (timestamp == ctx.getCurrentKey() + 63001L) {
                mapState.clear();
                return;
            }

            ArrayList<UrlViewCountWindow> arrayList = new ArrayList<>();
            Iterator<Map.Entry<String, UrlViewCountWindow>> iterator = mapState.iterator();
            while (iterator.hasNext()) {
                UrlViewCountWindow urlViewCountWindow = iterator.next().getValue();
                arrayList.add(urlViewCountWindow);
            }

            arrayList.sort(
                    new Comparator<UrlViewCountWindow>() {
                        @Override
                        public int compare(UrlViewCountWindow o1, UrlViewCountWindow o2) {
                            return o2.count.intValue() - o1.count.intValue();
                        }
                    }
            );
            StringBuilder result = new StringBuilder();
            result.append("=========================="+new Timestamp(timestamp -1L)+"==========================\n");
            for (int i = 0; i < Math.min(n,arrayList.size()); i++) {
                UrlViewCountWindow tmp = arrayList.get(i);
                result.append("NO. "+(i+1) + "  url:"+tmp.url +",被浏览次数："+tmp.count+"\n");
            }
            result.append("========================================================\n");

            out.collect(result.toString());
        }
    }

    public static class WindowResult extends ProcessWindowFunction<Long, UrlViewCountWindow, String, TimeWindow> {
        @Override
        public void process(String key, ProcessWindowFunction<Long, UrlViewCountWindow, String, TimeWindow>.Context context, Iterable<Long> elements, Collector<UrlViewCountWindow> out) throws Exception {
            Long cnt = elements.iterator().next();
            out.collect(new UrlViewCountWindow(
                    key,
                    cnt,
                    context.window().getStart(),
                    context.window().getEnd()
            ));
        }
    }

    public static class CountAgg implements AggregateFunction<ApacheLog, Long, Long> {
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(ApacheLog value, Long accumulator) {
            accumulator += 1L;
            return accumulator;
        }

        @Override
        public Long getResult(Long accumulator) {
            return accumulator;
        }

        @Override
        public Long merge(Long a, Long b) {
            return a + b;
        }
    }


}
